package com.silky.agent;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

public class DistributedApp {
    private static final String NAMESRV_ADDR = "localhost:9876"; // 替换为实际的Broker地址
    private static final String STARTUP_TOPIC = "StartupTopic";
    private static final String SHUTDOWN_TOPIC = "ShutdownTopic";
    private static final String STARTUP_GROUP = "StartupConsumerGroup";
    private static final String SHUTDOWN_GROUP = "ShutdownConsumerGroup";

    private DefaultMQPushConsumer startupConsumer;
    private DefaultMQPushConsumer shutdownConsumer;
    private DefaultMQProducer startupProducer;
    private DefaultMQProducer shutdownProducer;

    public DistributedApp() throws MQClientException {
        initializeConsumers();
        initializeProducers();
    }

    private void initializeConsumers() throws MQClientException {
        // 初始化启动消息消费者
        startupConsumer = new DefaultMQPushConsumer(STARTUP_GROUP);
        startupConsumer.setNamesrvAddr(NAMESRV_ADDR);
        startupConsumer.subscribe(STARTUP_TOPIC, "*");
        startupConsumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    String message = new String(msg.getBody());
                    if ("START".equals(message)) {
                        System.out.println(" [x] Received startup message: '" + message + "'");
                        // 执行启动逻辑
                        startApplication();
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        // 初始化停止消息消费者
        shutdownConsumer = new DefaultMQPushConsumer(SHUTDOWN_GROUP);
        shutdownConsumer.setNamesrvAddr(NAMESRV_ADDR);
        shutdownConsumer.subscribe(SHUTDOWN_TOPIC, "*");
        shutdownConsumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    String message = new String(msg.getBody());
                    if ("SHUTDOWN".equals(message)) {
                        System.out.println(" [x] Received shutdown message: '" + message + "'");
                        // 执行停止逻辑
                        stopApplication();
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
    }

    private void initializeProducers() throws MQClientException {
        // 初始化启动消息生产者
        startupProducer = new DefaultMQProducer("StartupProducerGroup");
        startupProducer.setNamesrvAddr(NAMESRV_ADDR);
        startupProducer.start();

        // 初始化停止消息生产者
        shutdownProducer = new DefaultMQProducer("ShutdownProducerGroup");
        shutdownProducer.setNamesrvAddr(NAMESRV_ADDR);
        shutdownProducer.start();
    }

    public void start() throws MQClientException, InterruptedException {
        startupConsumer.start();
        shutdownConsumer.start();
        System.out.printf("Consumers started.%n");

        // 发送启动消息
        sendStartupMessage();
    }

    public void stop() throws MQClientException, InterruptedException {
        // 发送停止消息
        sendShutdownMessage();

        // 关闭消费者和生产者
        shutdownConsumer.shutdown();
        startupConsumer.shutdown();
        shutdownProducer.shutdown();
        startupProducer.shutdown();
    }

    private void sendStartupMessage() throws MQClientException, InterruptedException {
        byte[] body = "START".getBytes();
        Message msg = new Message(STARTUP_TOPIC, body);
        startupProducer.send(msg);
        System.out.println("Sent startup message.");
    }

    private void sendShutdownMessage() throws MQClientException, InterruptedException {
        byte[] body = "SHUTDOWN".getBytes();
        Message msg = new Message(SHUTDOWN_TOPIC, body);
        shutdownProducer.send(msg);
        System.out.println("Sent shutdown message.");
    }

    private void startApplication() {
        // 执行具体的启动逻辑
        System.out.println("Starting application...");
    }

    private void stopApplication() {
        // 执行具体的停止逻辑
        System.out.println("Stopping application...");
    }

    public static void main(String[] args) throws MQClientException, InterruptedException {
        DistributedApp app = new DistributedApp();
        app.start();
        // 等待一段时间
        Thread.sleep(5000);
        app.stop();
    }
}

